#ifndef MQCLIENT_MQ_CHANNEL_HPP_
#define MQCLIENT_MQ_CHANNEL_HPP_
#include "muduo/proto/codec.h"
#include "../mqcomm/proto/mq_protocol.pb.h"
#include "../mqcomm/mq_helper.hpp"
#include "mq_consumer.hpp"
#include <condition_variable>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>

namespace mqclient
{

using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
using basicCommonResponsePtr = std::shared_ptr<proto::protocol::basicCommonResponse>;
using basicConsumeResponsePtr = std::shared_ptr<proto::protocol::basicConsumeResponse>;

class Channel
{
public:
    Channel(const muduo::net::TcpConnectionPtr& conn, const ProtobufCodecPtr& codec)
        : channel_id(helper::UUIDHelper::uuid()), conn_(conn), codec_(codec)
    {}
    ~Channel() { basicCancel(); }   // 信道关闭的时候取消订阅

    std::string channelId() { return channel_id; }

    bool openChannel()
    {
        std::string rid = helper::UUIDHelper::uuid();
        proto::protocol::openChannelRequest req;
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        codec_->send(conn_, req);
        auto resp = waitResponse(rid);
        return resp->ok();
    }
    void closeChannel()
    {
        std::string rid = helper::UUIDHelper::uuid();
        proto::protocol::closeChannelRequest req;
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        codec_->send(conn_, req);
        auto resp = waitResponse(rid);
    }

    // 声明&删除交换机
    bool declareExchange(const std::string& name,
                proto::message::ExchangeType type, 
                bool durable, bool auto_delete, 
                google::protobuf::Map<std::string, std::string>& args)
    {
        // 1. 构造请求对象
        proto::protocol::declareExchangeRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_exchange_name(name);
        req.set_exchange_type(type);
        req.set_durable(durable);
        req.set_auto_delete(auto_delete);
        req.mutable_args()->swap(args);
        // 2. 向服务器发送请求
        codec_->send(conn_, req);
        // 3. 等待服务器响应(核心)
        //    muduo库的网络通信是异步的, send返回时, 只是将数据发送到muduo底层的发送缓冲区
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    void deleteExchange(const std::string& name) 
    {
        proto::protocol::deleteExchangeRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_exchange_name(name);
        waitResponse(rid);
    }

    // 创建&删除队列
    bool declareQueue(const std::string name, bool durable, 
                bool exclusive, bool auto_delete,
                google::protobuf::Map<std::string, std::string>& args)
    {
        proto::protocol::declareQueueRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_queue_name(name);
        req.set_exclusive(exclusive);
        req.set_durable(durable);
        req.set_auto_delete(auto_delete);
        req.mutable_args()->swap(args);

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    void deleteQueue(const std::string& name)
    {
        proto::protocol::deleteQueueRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_queue_name(name);
        waitResponse(rid);     
    }

    // 交换机-队列绑定&解绑
    bool queueBind(const std::string& ename, const std::string& qname,
              const std::string& binding_key)
    {
        proto::protocol::queueBindRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        req.set_binding_key(binding_key);

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    void queueUnBind(const std::string &ename, const std::string &qname)
    {
        proto::protocol::queueBindRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_exchange_name(ename);
        req.set_queue_name(qname);
        waitResponse(rid);
    }

    // 发布&确认消息
    bool basicPublish(const std::string& ename, 
                    proto::message::BasicProperties* properties,
                    const std::string& body)
    {
        proto::protocol::basicPublishRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_exchange_name(ename);
        req.set_body(body);
        if(properties) {
            req.mutable_properties()->set_id(properties->id());
            req.mutable_properties()->set_delivery_mode(properties->delivery_mode());
            req.mutable_properties()->set_routing_key(properties->routing_key());
        }

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }
    bool basicAck(const std::string &msg_id)
    {
        if (consumer_.get() == nullptr) {
            LOG_ERROR << "The consumer cannot be found when the message basicAck.\n";
            return false;
        }
        proto::protocol::basicAckRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_queue_name(consumer_->qname_);
        req.set_message_id(msg_id);

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        return resp->ok();
    }

    // 添加&取消订阅
    bool basicConsume(const std::string& consumer_tag,
                            const std::string& qname, bool auto_ack,
                            const ConsumerCallback& callback)
    {
        if (consumer_.get()) {
            LOG_ERROR << "The current channel has been consume to other queue messages.\n";
            return false;
        }
        proto::protocol::basicConsumeRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_consumer_tag(consumer_tag);
        req.set_queue_name(qname);
        req.set_auto_ack(auto_ack);

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        if(resp->ok() == false) {
            LOG_DEBUG << "add consume fail\n";
            return false;
        }
        consumer_ = std::make_shared<Consumer>(consumer_tag, qname, auto_ack, callback);
        return true;
    }
    void basicCancel()
    {
        // 预处理(该信道没有订阅者直接返回)
        if (consumer_.get() == nullptr) return;

        proto::protocol::basicCancelRequest req;
        std::string rid = helper::UUIDHelper::uuid();
        req.set_rid(rid);
        req.set_channel_id(channel_id);
        req.set_consumer_tag(consumer_->tag_);
        req.set_queue_name(consumer_->qname_);

        codec_->send(conn_, req);
        basicCommonResponsePtr resp = waitResponse(rid);
        consumer_.reset();      // 重置订阅者
    }

// 向连接模块提供接口
public:     
    // 连接收到基础响应后, 向hash_map中添加响应
    void putBasicResponse(const basicCommonResponsePtr& resp)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        basic_resp_.emplace(resp->rid(), resp);
        // 已经收到响应, 可以通知条件变量去检测
        resp_cv_.notify_all();      
    }
    // 连接收到消息推送请求后, 通过信道找到对应的消费者对象, 再通过消费者对象内的回调函数进行消息处理
    void consume(const basicConsumeResponsePtr& resp)
    {
        if(consumer_.get() == nullptr) {
            LOG_ERROR << "The channel has no consumer.\n";
            return;
        }
        if(consumer_->tag_ != resp->consumer_tag()) {
            LOG_ERROR << "The consumer does not belong to the current channel.\n";
            return;   
        }
        consumer_->callback_(consumer_->tag_, resp->mutable_properties(), resp->body());
    }
private:
    basicCommonResponsePtr waitResponse(const std::string& rid)
    {
        std::unique_lock<std::mutex> lock(mtx_);
        // resp_cv 条件不满足则等待, 直到收到响应
        resp_cv_.wait(lock, [&rid, this](){ 
            return basic_resp_.find(rid) != basic_resp_.end(); 
        });
        basicCommonResponsePtr resp = basic_resp_[rid];
        basic_resp_.erase(rid);     // 移除已经收到的响应
        return resp;
    }
private:
    std::string channel_id;
    muduo::net::TcpConnectionPtr conn_;
    ProtobufCodecPtr codec_;
    ConsumerPtr consumer_; 
    std::mutex mtx_;
    std::condition_variable resp_cv_;
    std::unordered_map<std::string, basicCommonResponsePtr> basic_resp_;    // <rid, basicCommonResponse>

};

using ChannelPtr = std::shared_ptr<Channel>;

class ChannelManager
{
public:
    ChannelManager() = default;
    ~ChannelManager() = default;
    ChannelPtr create(const muduo::net::TcpConnectionPtr& conn,
                      const ProtobufCodecPtr codec)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        auto channel = std::make_shared<Channel>(conn, codec);
        channels_.emplace(channel->channelId(), channel);
        return channel;
    } 
    void remove(const std::string channel_id)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        channels_.erase(channel_id);
    }
    ChannelPtr getChannel(const std::string channel_id)
    {
        std::lock_guard<std::mutex> lock(mtx_);
        return channels_.find(channel_id) != channels_.end() ?
               channels_[channel_id] :
               ChannelPtr();
    }
private:
    std::mutex mtx_;
    std::unordered_map<std::string, ChannelPtr> channels_;  // <channel_id, channel>
};

using ChannelManagerPtr = std::shared_ptr<ChannelManager>;

}

#endif